package com.danbay.framework.kafka.config;

import com.danbay.framework.exception.MessageCodeException;
import com.danbay.framework.kafka.*;
import com.danbay.framework.kafka.coap.CoAPMessage;
import com.danbay.framework.protocol.command.AbstractCommandHandler;
import com.danbay.framework.protocol.command.ProtocolMessage;
import com.danbay.framework.protocol.command.codec.ICmdDataDecoder;
import com.danbay.framework.protocol.command.codec.ICmdDataEncoder;
import com.danbay.framework.utils.ByteUtils;
import com.danbay.framework.utils.StringUtils;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.eclipse.californium.core.coap.CoAP;
import org.eclipse.californium.core.coap.Message;
import org.eclipse.californium.core.coap.Response;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.kafka.annotation.KafkaListener;

import java.util.List;

/**
 * @author SJG
 *         2017/11/16.
 */
@ConfigurationProperties
public class AutoKafkaListener extends AbstractCommandHandler {
    private static final String ACK_CONTENT = "ACK";
    /**
     * 有些处理回复ACK的内容不需要做处理.
     */
    private static final String NOT_NEED_ACK_CMD = "/ctrl/get_status_kwh,/ctrl/get_relays";
    @Value("${app.kafka.owner.topic-name:unnamed}")
    private String ownerTopicName;

    private final ICmdDataDecoder dataDecoder;
    private final ICmdDataEncoder dataEncoder;
    private ObjectMapper objectMapper;
    @Autowired
    private KafkaClient kafkaClient;

    public AutoKafkaListener(ICmdDataDecoder dataDecoder, ICmdDataEncoder dataEncoder) {
        this.dataDecoder = dataDecoder;
        this.dataEncoder = dataEncoder;
        this.objectMapper = new ObjectMapper();
    }

    /**
     * 监听来自access的kafka消息
     *
     * @param record 消息
     */
    @KafkaListener
    public void autoListenKafkaMessageFromAccess(ConsumerRecord<String, AccessPackage<Message>> record) {
        if (record == null || record.value() == null) {
            logger.info("message record is null,ignore...");
            return;
        }

        AccessPackage<Message> accessPackage = record.value();
        AccessHeader header = accessPackage.getHeader();
        //kafka ack,不处理
        if (header.getType() == AccessPackage.MESSAGE_TYPE_RESPONSE) {
            logger.info("kafka ack message,ignore...");
            return;
        }
        //自己发给自己的不处理
//        if(ownerTopicName!=null && ownerTopicName.contains(header.getTopicName())){
//            logger.info("kafka send to self message,ignore...");
//            return;
//        }

        try {
            //response topic
            AccessKafkaTopic kafkaTopic = kafkaClient.responseTopic(header.getTopicName());

            //取得CoAP内容
            Message message = accessPackage.getData();
            //取得payload和uri query
            String encoding = "utf8";
            String msg = new String(message.getPayload(), encoding);
            boolean noPayload = StringUtils.isBlank(msg);
            String path = StringUtils.join(message.getOptions().getUriPath(), "/");
            //如果没传uri path，则从redis中取得下发请求时mid对应的uri path
            if (StringUtils.isBlank(path)) {
                path = kafkaTopic.getCoAPUriPathByMid(message.getMID());
            } else {
                path = "/" + path;
            }

            //ACK消息，不处理
            if (message.getType() == CoAP.Type.ACK) {
                //不需要回ack的情况：1内容为ack,2命令是/ctrl/get_status_kwh
                if (NOT_NEED_ACK_CMD.contains(path)) {
                    return;
                }
            }

            //处理错误码
            if (message instanceof Response) {
                if (((Response) message).getCode().value > CoAP.ResponseCode.CONTINUE.value) {
                    logger.warn("receive response error");
                    return;
                }
            }

            //回应ack
            kafkaTopic.ack(header.getSequence());

            List<String> uriQuery = message.getOptions().getUriQuery();
            //获取原请求的messageId
            String messageId = kafkaTopic.getKafkaMessageIdByMid(message.getMID());
            //如果有对应的messageId则添加到回调方法参数中去
            if (messageId != null) {
                uriQuery.add("messageId=" + messageId);
            }

            CoAPMessage coAPMessage = new CoAPMessage();
            coAPMessage.setPayload(msg);
            coAPMessage.setUriQuery(uriQuery);
            //合并payload和uri query
            msg = objectMapper.writeValueAsString(coAPMessage);

            //取请求的token
            Long srcToken = null;
            if (message.getToken() != null && message.getToken().length != 0) {
                //token作为事务功能，无事务可不用token
                srcToken = ByteUtils.bytes2Long(message.getToken());
            }

            //响应业务处理结果
            AccessKafkaTopic requestTopic = kafkaClient.requestTopic(header.getTopicName());
            //取出uri query中的dev_id参数回传
            String devIdQuery = uriQuery.stream().filter(s -> s.startsWith("dev_id=")).findFirst().orElse("");
            Object result;
            try {
                //保持oauth token到当前线程
                KafkaContext.putToken(header.getOauthToken());
                KafkaContext.putTopic(header.getTopicName());
                //处理命令业务
                result = invokeCmd(
                        new ProtocolMessage(getCommandMethodTypeName(message.getType(), noPayload) + path, msg),
                        null,
                        dataDecoder,
                        dataEncoder);
            } catch (MessageCodeException e) {
                //回复调用结果
                ackData(message, path, devIdQuery, requestTopic, srcToken, null, true,e.getCode());
                throw e;
            }catch (Exception e) {
                //回复调用结果
                ackData(message, path, devIdQuery, requestTopic, srcToken, null, true);
                throw e;
            }finally {
                //释放
                KafkaContext.popToken();
                KafkaContext.popTopic();
            }

            //回复调用结果
            ackData(message, path, devIdQuery, requestTopic, srcToken, result, false);
        } catch (Exception e) {
            logger.error("invokeCmd error!", e);
        }

    }

    /**
     * 回应调用结果消息
     *
     * @param message      message
     * @param path         path
     * @param devIdQuery   devIdQuery
     * @param requestTopic requestTopic
     * @param srcToken     srcToken
     * @param result       result
     * @param isError      isError
     * @param <T>          result type
     */
    private <T> void ackData(Message message, String path, String devIdQuery, AccessKafkaTopic requestTopic, Long srcToken, T result, boolean isError) {
        ackData(message,path,devIdQuery,requestTopic,srcToken,result,isError,400);
    }

    /**
     * 回应调用结果消息
     *
     * @param message      message
     * @param path         path
     * @param devIdQuery   devIdQuery
     * @param requestTopic requestTopic
     * @param srcToken     srcToken
     * @param result       result
     * @param isError      isError
     * @param <T>          result type
     */
    private <T> void ackData(Message message, String path, String devIdQuery, AccessKafkaTopic requestTopic, Long srcToken, T result, boolean isError,int code) {
        //NON、ACK、类型消息，不需要回复
        if (message.getType() == CoAP.Type.NON || message.getType() == CoAP.Type.ACK || message.getType() == CoAP.Type.RST) {
            return;
        }
        if (StringUtils.isNotBlank(devIdQuery)) {
            path = path + "?" + devIdQuery;
        }
        if (isError) {
            //回应错误,指定错误码
            if (code == 404){
                requestTopic.ackBadRequest(CoAP.ResponseCode.NOT_FOUND,path
                        ,message.getMID(),srcToken);
                return;
            }
            //默认返回400
            requestTopic.ackBadRequest(
                    path,
                    message.getMID(),
                    srcToken
            );
        } else {
            //回应正确数据
            requestTopic.ackData(
                    path,
                    message.getMID(),
                    srcToken,
                    result
            );
        }
    }

    /**
     * 取命令对应controller的action的方法的RequestMethod名称
     *
     * @param type CoAP命令类型
     * @return 命令类型名
     */
    private String getCommandMethodTypeName(CoAP.Type type, boolean noPayload) {
        if (type == CoAP.Type.ACK) {
            //payload无数据，只是回应状态码
            if (noPayload) {
                return "get:";
            }
            //有数据提交
            return "put:";
        } else {
            return "post:";
        }
    }

    @Override
    public ICmdDataDecoder getDataDecoder() {
        return null;
    }

    @Override
    public ICmdDataEncoder getDataEncoder() {
        return null;
    }

    @Override
    public void init() {

    }
}
